package com.nokia.position;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;
import java.util.TimeZone;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;

public class PositionServiceUsageUSA2Mapper
		extends
		Mapper<LongWritable, ProtobufWritable<Position>, Text, PositionServiceUsageUSA2Writable> {

	public static final Logger LOGGER = Logger
			.getLogger(PositionServiceUsageUSA2Mapper.class.getName());

	Hashtable<String, String> operatorCountryMap;
	Hashtable<String, String> operatorNameMap;

	@Override
	protected void map(LongWritable key, ProtobufWritable<Position> value,
			Context context) throws IOException, InterruptedException {
		// IntWritable One = new IntWritable(1);

		// per date/hour, per country, per network type, per platform, per
		// protocol, per
		// used positioning, per client name
		Position position = value.get();
		
		PositionServiceUsageUSA2Writable psw = new PositionServiceUsageUSA2Writable(1);

		// Operator Name
		String MccMnc = (position.hasServingMccSystemID() ? position
				.getServingMccSystemID() : "")
				+ "\t"
				+ (position.hasServingMncNetworkID() ? position
						.getServingMncNetworkID() : "");
		String operatorName = operatorNameMap.get(MccMnc);
		if (operatorName == null) {
			operatorName = "";
		}

		// Country
		Integer MccInt = (position.hasServingMccSystemID() ? position
				.getServingMccSystemID() : null);

		String country = null;
		if (MccInt != null) {
			country = operatorCountryMap.get(MccInt.toString());
		}
		if (country == null) {
			country = "";
		}

		// Platform type
		Integer platformType = (position.hasClientType() ? position
				.getClientType() : null);

		// Protocol type
		Integer protocolType = (position.hasProtocolType() ? position
				.getProtocolType() : null);
                
		// Manufacture
		Integer manufacture = (position.hasManufacturer() ? position
				.getManufacturer() : null);

		Date dateObj = new Date(position.getCoordinatesTimestamp());
		SimpleDateFormat parser = new SimpleDateFormat("yyyy-MM-dd\tHH");
		parser.setTimeZone(TimeZone.getTimeZone("UTC"));
		String partitionDate = parser.format(dateObj);

		context.write(
				new Text((partitionDate + "\t" +  MccMnc
						+ "\t" + country + "\t"
                                                + (manufacture != null ? manufacture : "") + "\t"
						+ (platformType != null ? platformType : "") + "\t"
						+ (protocolType != null ? protocolType : "") + "\t"
                                                + operatorName + "\t" 
						+ (position.hasClientName() ? position.getClientName() : ""))),
				psw);
	}

	public void setup(Context context) {

		operatorCountryMap = new Hashtable<String, String>();
		operatorNameMap = new Hashtable<String, String>();
        
		try {
			Path[] uris = DistributedCache.getLocalCacheFiles(context
					.getConfiguration());
			for (int i = 0; i < uris.length; i++) {
                                if (uris[i].toString().contains("mcc2country")) {
					try {
						BufferedReader operatorCountryMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readOperatorCountryMapCache(operatorCountryMapReader,
								this.operatorCountryMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				} else if (uris[i].toString().contains("MccMncOperatorCountry")) {
					try {
						BufferedReader operatorNameMapReader = new BufferedReader(
								new FileReader(uris[i].toString()));
						readOperatorNameMapCache(operatorNameMapReader,
								this.operatorNameMap);
					} catch (FileNotFoundException e1) {
						e1.printStackTrace();
					}
				}
			}
		} catch (IOException e2) {
			e2.printStackTrace();
		}
	}

	public static void readOperatorCountryMapCache(
			BufferedReader operatorCountryMapReader,
			Hashtable<String, String> dim) throws IOException {

		// 1\t412\tAF

		// Key is 2nd column, value (country) is 3rd

		String line;
		while ((line = operatorCountryMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[1], new String(fields[2]));
		}
	}

	public static void readOperatorNameMapCache(
			BufferedReader operatorNameMapReader, Hashtable<String, String> dim)
			throws IOException {

		// 1\t412\tAF
		// 1504\t310\t970\tGlobalstar USA\tUnited States

		// Key is 2nd and 3rd column, value (operator id) is 1st

		String line;
		while ((line = operatorNameMapReader.readLine()) != null) {
			String[] fields = line.toString().split("\t", -1);
			dim.put(fields[1] + "\t" + fields[2], new String(fields[0]));
		}
	}
}
